package hyl.base.mq.mqtt;

import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import hyl.base.mq.BaseMQ;
import hyl.core.Amy;
import hyl.core.MyFun;
import hyl.core.conf.MyConfig;
import hyl.core.info.CMsg;

public class MyMQTT extends BaseMQ {
	// static final int M_广播=1;
	public static final int M_一对一 = 2;
	/** 一对多,排队轮流谈 */
	public static final int M_轮发 = 3;
	/** 一对多组播 */
	public static final int M_群发 = 4;
	/** 一对多组播-多个特定主题 */
	// public static final int M_群发精确主题 = 5;
	/** 一对多组播相似主题 */
	// public static final int M_群发模糊主题 = 6;
	static final String file_conf = "mq/mqtt.conf";
	public static final Logger Log = LoggerFactory.getLogger(MyMQTT.class);
	MqttConnectOptions options_sub;
	MqttClient mqttClient;

	private MyMQTT() {
	}

	boolean _应答方式 = true;

	private MyMQTT(MqttClient client, MqttConnectOptions options) {
		mqttClient = client;
		options_sub = options;
	}

	public static MyMQTT getInstance(String clientId) {
		// 获取配置信息
		Map<String, String> mqttProperty = MyConfig.loadIni(file_conf, Amy.charset);
		// MQTT的连接设置
		MqttConnectOptions options_sub = new MqttConnectOptions();
		// setCleanSession =true:临时会话  断开后 清除之前订阅的主题
		// =false 持久会话 断开后 保留之前订阅的主题 ,速度快,但比较危险
		// 如果一个MqttClient客户端订阅过多个主题, 下一次订阅时,会订阅之前订阅过的所有主题,
		// 最好一个MqttClient抽象客户端用一类,不要订阅不同的主题
		options_sub.setCleanSession(false);
		// 设置连接的用户名
		options_sub.setUserName(mqttProperty.get("MQTT.USERNAME"));
		// 设置连接的密码
		options_sub.setPassword(mqttProperty.get("MQTT.PASSWORD").toCharArray());
		// 设置超时时间 单位为秒
		options_sub.setConnectionTimeout(10);
		// 设置会话心跳时间 单位为秒 服务器会每隔1.5*20秒的时间向客户端发送个消息判断客户端是否在线，
		//但这个方法并没有重连的机制
		options_sub.setKeepAliveInterval(20);
//		设置持久化方式MemoryPersistence设置clientid的保存形式，默认为以内存保存
		// MemoryPersistence memoryPersistence = new MemoryPersistence();
		String host = mqttProperty.get("MQTT.TCP.HOST");
		// HOST_MQ为主机名，clientid即连接MQTT的客户端ID，一般以唯一标识符表示，
		if (MyFun.isEmpty(host) || MyFun.isEmpty(clientId))
			return null;
		// 设置遗言消息 异常断开前的留言
		// options_sub.setWill(topic, "close".getBytes(), 2, true);
		try {
			MqttClient mqttClient = new MqttClient(host, clientId, new MemoryPersistence());
			return new MyMQTT(mqttClient, options_sub);
		} catch (MqttException e) {
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * mqtt重连
	 */
	public MyMQTT reconnect() {
		try {
			if (null != mqttClient && !(mqttClient.isConnected())) {
				mqttClient.reconnect();
				Log.info("尝试连接MQTT");
			}
		} catch (MqttException e) {
			Log.error("重连MQTT失败:{}", e.toString());
		}
		return this;
	}

	/**
	 * 连接前 ,bind回调函数
	 * 
	 * @param callback
	 * @return
	 */
	MyMQTT connect() {
		try {
			// 订阅topic定义
			// String[] topics = new String[] { topic_online, topic_will, topic_dev };
			// 设置回调
			mqttClient.setCallback(new MqttCallback() {
				public void connectionLost(Throwable cause) {
					// 连接丢失后，一般在这里面进行重连
					Log.error("连接断开,请重连");
				}

				public void deliveryComplete(IMqttDeliveryToken token) {
					Log.info("传送完成---------" + token.isComplete());
				}

				public void messageArrived(String topic, MqttMessage message) throws Exception {
					if (_callback != null) {
						CMsg msg = new CMsg();
						msg.s主题 = topic;
						msg.bs内容 = message.getPayload();
						if (_callback.success(msg)) {
						}
					}
				}
			});
			// 连接mqtt服务器broker
			// 这是一个阻塞方法，在连接完成后返回
			if (null != mqttClient && !(mqttClient.isConnected())) {
				// mqttClient.reconnect();
				mqttClient.connect(options_sub);
			}
			// System.out.println("连接状态:" + mqttClient.isConnected());
		} catch (MqttException e) {

			if (_callback != null) {
				_callback.error(null, e);
			} else {
				Log.error("=======连接失败:{}==============", e);
			}

		}
		return this;
	}

	/**
	 * 连接后才能发送消息
	 * 
	 * @param qos
	 */
	public void publishMessage(String 发送主题, Integer qos, byte[] 发送内容) {

		MqttMessage mqttMessage = new MqttMessage();
		if (qos != null)
			mqttMessage.setQos(qos);// 指定发送确认模式
		// setRetained :消息引擎是否应保留发布消息。默认false 不保留 ,不保留,后启动的接收方收不到消息
		mqttMessage.setRetained(true);
		mqttMessage.setPayload(发送内容);// 设置发送内容

		if (发送主题 != null)// 重改
			_topic = 发送主题;

		MqttTopic topic = mqttClient.getTopic(_topic);
		CMsg result = new CMsg();
		result.s主题 = _topic;// 队列名称
		result._m属性.put("key", _key);// 分组
		result.bs内容 = 发送内容;// 内容类型
		if (null != topic) {
			try {
				/**
				 * MqttDeliveryToken提供跟踪邮件传递进度的机制。
				 *
				 * 用于跟踪以非阻塞方式执行发布时消息的传递进度（在后台运行）
				 */
				topic.publish(mqttMessage);
				if (_callback != null) { // 如果发送成功返回发送状态
					_callback.success(result);
				}
			} catch (MqttException e) {
				if (_callback != null) { // 如果发送失败返回异常状态
					_callback.error(result, e);
				} else {
					e.printStackTrace();
				}
			}
		}
	}

	/**
	 * 连接后订阅消息
	 * 
	 * @param pubTopic
	 * @param qos
	 */
	void consumerMessage(int Qos) {
		String[] topics = _topic.split(",");
		int len = topics.length;
		int[] Qoss = new int[len];
		for (int i = 0; i < len; i++) {
			Qoss[i] = Qos;
		}
		consumerMessage(topics, Qoss);
	}

	/**
	 * 连接后订阅消息
	 * 
	 * @param topics 可以是多个主题
	 * @param qos
	 */
	void consumerMessage(String[] topics, int[] Qos) {
		try {
			// 订阅消息
			mqttClient.subscribe(topics, Qos);
		} catch (Exception e) {
			Log.error(null, "消费消息异常{}:{}:", StringUtils.join(topics, ","), e.getMessage());
		}
	}

	@Override
	public void close() {
		try {
			mqttClient.disconnect();
			mqttClient.close();
		} catch (MqttException e) {
			e.printStackTrace();
		}
	}

	// 清空主题
	public void remove() {
		if (null != mqttClient && !mqttClient.isConnected()) {
			try {
				String[] title = _topic.split(",");
				mqttClient.unsubscribe(title);
			} catch (MqttException e) {
				e.printStackTrace();
			}
		} else {
			System.out.println("mqttClient is error");
		}
	}

	@Override
	public void send(String 数据标签, byte[] msg) {
		publishMessage(数据标签, 1, msg);
	}

	@Override
	public void receive() {
		connect();
		consumerMessage(1);
	}
}
